package csp

import (
	"sync"
	"time"
)

/*
	发布订阅模式 Pub/Sub
	传统生产者和消费者模型中，是将消息发送到一个队列中，而发布订阅模型则是将消息发布给一个主题
*/

type (
	subscriber chan interface{}         //订阅者为一个管道
	topicFun   func(v interface{}) bool //订阅者处理消息的函数，bool方便判断是否处理成功，这里不作retry实现
)

//发布者对象
type Publisher struct {
	m           sync.RWMutex            //读写锁，因为publisher允许被并发使用
	buffer      int                     //订阅队列的缓存大小
	timeout     time.Duration           //发布超时时间
	subscribers map[subscriber]topicFun //订阅者消息
}

//构建一个发布者对象，可以设置发布超时时间和缓存队列的长度
func NewPublisher(publishTimeout time.Duration,buffer int) *Publisher {
	return &Publisher{
		buffer: buffer,
		timeout: publishTimeout,
		subscribers: make(map[subscriber]topicFun),
	}
}

//添加一个新的订阅者，订阅全部主题
func (p *Publisher) Subscribe()  chan interface{}{
	return p.SubscribeTopic(nil)
}

//添加一个新的订阅者，订阅过滤器筛选后的主题
func (p *Publisher) SubscribeTopic(topic topicFun) chan interface{}{
	ch := make(chan interface{},p.buffer)
	p.m.Lock()
	p.subscribers[ch] = topic
	p.m.Unlock()
	return ch
}

//退出订阅
func (p *Publisher Evict(sub chan interface{})  {
	p.m.Lock()
	defer p.m.Unlock()
	delete(p.subscribers,sub)
	close(sub)
}

//发布一个主题
func (p *Publisher) Pubnlish(v interface{})  {
	p.m.RLock()
	defer p.m.RUnlock()
	
	var wg sync.WaitGroup
	for sub,topic := range p.subscribers {
		wg.Add(1)
		go p.sendTopic(sub,topic,v,&wg)
	}
}

//关闭发布者对象，同时关闭所有的订阅者管道
func (p *Publisher) Close()  {
	p.m.Lock()
	defer p.m.Unlock()
	
	for sub := range p.subscribers {
		delete(p.subscribers,sub)
		close(sub)
	}
}


//发送主题。可以容忍一定的超时
func (p *Publisher) sendTopic(sub subscriber,topic topicFun,v interface{},wg *sync.WaitGroup)  {
	defer wg.Done()
	if topic != nil && !topic(v) {
		return
	}

	select {
	case sub <- v:
	case <-time.After(p.timeout):
	}
}